package com.lichao.cdc;

import com.lichao.pojo.BehaviorFull;
import com.lichao.pojo.ClickBehaviorFull;
import com.lichao.pojo.ClickSource1;
import com.lichao.pojo.MysqlSource;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

// 广播流
public class BroadcastStateExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取用户行为事件流
        DataStreamSource<ClickBehaviorFull> actionStream = env.addSource(new ClickSource1());
        // 定义行为模式流，代表了要检测的标准
        DataStreamSource<BehaviorFull> patternStream = env.addSource(new MysqlSource());
        // 定义广播状态的描述器，创建广播流
        MapStateDescriptor<String, BehaviorFull> bcStateDescriptor =
                new MapStateDescriptor<>("behaviorFull", Types.STRING, Types.POJO(BehaviorFull.class));

        BroadcastStream<BehaviorFull> bcPatterns = patternStream.broadcast(bcStateDescriptor);

        // 将事件流和广播流连接起来，进行处理
        DataStream<Tuple2<String, ClickBehaviorFull>> matches = actionStream.keyBy(ClickBehaviorFull::getClientId)
                .connect(bcPatterns)
                .process(new PatternEvaluator());
        matches.print();
        env.execute();
    }

    public static class PatternEvaluator extends KeyedBroadcastProcessFunction<String, ClickBehaviorFull, BehaviorFull, Tuple2<String, ClickBehaviorFull>> {

        @Override
        public void processBroadcastElement(BehaviorFull pattern, Context ctx, Collector<Tuple2<String, ClickBehaviorFull>> out) throws Exception {
            BroadcastState<String, BehaviorFull> bcState = ctx.getBroadcastState(
                    new MapStateDescriptor<>("behaviorFull", Types.STRING, Types.POJO(BehaviorFull.class)));
            // 将广播状态更新为当前的 pattern
            System.out.println("更新检测 " + pattern);
            bcState.put(pattern.getClientId(), pattern);
        }

        @Override
        public void processElement(ClickBehaviorFull action, ReadOnlyContext ctx, Collector<Tuple2<String, ClickBehaviorFull>> out) throws Exception {
            BehaviorFull pattern = ctx.getBroadcastState(
                            new MapStateDescriptor<>("behaviorFull", Types.STRING, Types.POJO(BehaviorFull.class)))
                    .get(action.getClientId());

            System.out.println("数据源2" + action.getClientId());
            if (pattern != null) {
                if (pattern.getClientId().equals(action.getClientId())) {
                    out.collect(new Tuple2<>(ctx.getCurrentKey(), action));
                }
            }
        }
    }
}